/**
 * 
 */
package com.simple.datax.recv.impl;

import java.util.UUID;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.Connector;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.api.MessageProcessor;
import com.simple.datax.api.Receiver;
import com.simple.datax.api.SimpleException;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.conn.impl.ConnectorPoolManager;
import com.simple.datax.core.model.connpara.CommonConnPara;

/** 
 * @Title: ScheduledReceiverImpl.java 
 * @Package com.simple.datax.recv.impl 
 * @Description: Receiver 实现 
 * @author Joe Jin 
 * @date 2012-12-25 
 * @version V1.0 
 */
public class ScheduledReceiverImpl implements Receiver {
	private static final Logger logger = LoggerFactory
			.getLogger(ScheduledReceiverImpl.class);
	private MessageProcessor nextProcessor;
	private ConnectorPool connectorPool = ConnectorPoolManager.getInstance().getConnectorPool(ConnectorPoolManager.CONN_RECV, null, null);

	/* (non-Javadoc)
	 * @see com.simple.datax.api.Receiver#process(com.simple.datax.core.model.connpara.CommonConnPara)
	 */
	@Override
	public Runnable process(Connector connector, CommonConnPara para, int threshold) {
		/**
		 * 每次都new一个新对象，目前没有太大问题。如果实际运行时存在性能问题，
		 * 可将ProcessRunnable保存在Map中，以 Connector对象为key
		 */
		return new ProcessRunnable(connector, para, threshold);
//		logger.debug("Thread: "+ Thread.currentThread().getId() + " - ScheduledReceiverImpl.processor");
//		String message = null;
//		try {
//			// got connector then continue
//			if (connector != null) {
//				connector.connect();
//				// get message one by one
//				// the number of thread depends on setting in Spring 
////				while ((message = connector.getMessage()) != null) {
////					SimpleMessage msgObj = createSimpleMessage(para);
////					msgObj.setContent(message);
////					nextProcessor.process(msgObj);
////				}
//				SimpleMessage msgObj = createSimpleMessage(para);
//				nextProcessor.process(msgObj);
//			}
//		} catch (SimpleException e) {
//			logger.error("Reciver get message error.", e);
//		} finally {
//			if (connector != null) {
//				connector.disconnect();
//				try {
//					connector.release();
//				} catch (Exception e) {
//					logger.error("Connector release error.", e);
//				}
//			}
//		}
//		logger.debug("Reciver finished getting/handling message.");
	}
	
	private class ProcessRunnable implements Runnable {
		private Connector connector;
		private CommonConnPara para;
		private int threshold;
		public ProcessRunnable(Connector connector, CommonConnPara para, int threshold) {
			this.connector = connector;
			this.para = para;
			this.threshold = threshold;
		}

		@Override
		public void run() {
			//logger.debug("Thread: "+ Thread.currentThread().getId() + " - ScheduledReceiverImpl.processor");
			String message = null;
			try {
				// got connector then continue
				if (connector != null) {
					// get message one by one
					// the number of thread depends on setting in Spring 
					while (threshold>0) {
						if (((message = connector.getMessage()) != null)) {
							SimpleMessage msgObj = createSimpleMessage(para);
							msgObj.setContent(message);
							logger.debug(" Begin &&&&&&&& " + msgObj.toLog());
							nextProcessor.process(msgObj);
							logger.debug(" End &&&&&&&& " + msgObj.toLog());
							threshold--;
						} else {
							break;
						}
					}
				}
			} catch (SimpleException e) {
				logger.error("Reciver get message error.", e);
			} catch (ConnectorException e) {
				logger.error("Reciver get message error.", e);
			} catch (Exception e) {
				logger.error("Reciver get message error.", e);
			} finally {
				if (connector != null) {
					try {
						connectorPool.returnObject((BaseConnector)connector);
					} catch (Exception e) {
						logger.error("Connector release error.", e);
					}
				}
			}
//			logger.debug("Reciver finished getting/handling message.");
		}
		
	}
	
	/**
	 * Create SimpleMessage object with fields in CommonConnPara object
	 * @param CommonConnPara para
	 * @return
	 */
	private SimpleMessage createSimpleMessage(CommonConnPara para) {
		SimpleMessage ret = new SimpleMessage();
		ret.setCommunParam(para.getTransportPara());
		ret.setCommunType(para.getTransportType());
		ret.setEncoding(para.getEncoding());
		ret.setMsgType(para.getMsgType());
		ret.setMsgFormat(para.getMsgFormat());
		ret.setRecvRawMsgid(UUID.randomUUID().toString());
		return ret;
	}

	public MessageProcessor getNextProcessor() {
		return nextProcessor;
	}

	public void setNextProcessor(MessageProcessor nextProcessor) {
		this.nextProcessor = nextProcessor;
	}
}
